package com.xzx.flink.streamapi.state;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 每个用户的PV数，10秒输出一次
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/14 00:59
 */
public class ValueState_01_UserPV {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<ClickEvent> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }));
        stream.print();

        //10秒统计一次
        stream.keyBy(ClickEvent::getUser)
                .process(new KeyedProcessFunction<String, ClickEvent, String>() {
                    private ValueState<Long> countState;
                    private ValueState<Long> tsState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("pv", Types.LONG));
                        tsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts", Types.LONG));
                    }

                    @Override
                    public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        Long count = countState.value();
                        if (count == null) {
                            countState.update(1L);
                        } else {
                            countState.update(count + 1);
                        }

                        //达到10秒注册一个输出事件
                        if (tsState.value() == null) {
                            tsState.update(value.getTimestamp());
                            ctx.timerService().registerEventTimeTimer(value.getTimestamp() + 10000);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, ClickEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        // 事件时间到10秒了
                        out.collect("用户：" + ctx.getCurrentKey() + "的PV=" + countState.value());
                        tsState.clear();
                    }
                })
                .print();

        env.execute(ValueState_01_UserPV.class.getSimpleName());
    }
}
